package cn.com.idmy.mq;

import cn.com.idmy.mq.builder.MqMsgBuilder;
import cn.com.idmy.mq.builder.MqSendResultBuilder;
import cn.com.idmy.mq.coder.MqMsgEncoder;
import cn.com.idmy.mq.exception.MqMsgSendException;
import cn.com.idmy.mq.model.MqMsg;
import cn.com.idmy.mq.model.MqSendResult;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

import static java.text.MessageFormat.format;

@Slf4j
class RocketmqProducer implements IMqProducer {
    private final DefaultMQProducer producer;
    private final MqMsgEncoder encoder;
    private final MqMsgBuilder<Message> msgBuilder;
    private final MqSendResultBuilder<SendResult> sendResultBuilder;

    @SneakyThrows
    public RocketmqProducer(MqProps props, MqMsgEncoder encoder, MqMsgBuilder<Message> msgBuilder, MqSendResultBuilder<SendResult> sendResultBuilder) {
        this.encoder = encoder;
        this.msgBuilder = msgBuilder;
        this.sendResultBuilder = sendResultBuilder;
        AclClientRPCHook rpcHook = new AclClientRPCHook(new SessionCredentials(props.getPassword(), props.getUsername()));
        producer = new DefaultMQProducer(props.getNamespace(), format("{0}%{1}", props.getNamespace(), props.getProducerGroup()), rpcHook);
        producer.setNamesrvAddr(props.getServerAddress());
        producer.setSendMsgTimeout(props.getSendMsgTimeout());
        producer.setDefaultTopicQueueNums(props.getTopicQueueNumber());
    }

    @Override
    public void start() {
        try {
            producer.start();
        } catch (MQClientException e) {
            log.error("RocketMQ producer start error:" + e.getErrorMessage(), e);
        }
    }

    @Override
    public MqSendResult send(MqMsg msg) {
        try {
            Message rocketmq = msgBuilder.build(msg, encoder);
            rocketmq.setWaitStoreMsgOK(true);

            SendResult result = producer.send(rocketmq);
            return sendResultBuilder.build(result);
        } catch (Exception e) {
            throw new MqMsgSendException(msg.getTopic(), e);
        }
    }

    @Override
    public MqSendResult sendOrderly(MqMsg msg) {
        try {
            Message rocketmq = msgBuilder.build(msg, encoder);
            rocketmq.setWaitStoreMsgOK(true);

            SendResult result = producer.send(rocketmq, new MqMsgSelector(), msg.getShardingKey());
            return sendResultBuilder.build(result);
        } catch (Exception e) {
            throw new MqMsgSendException(msg.getTopic(), e);
        }
    }

    @Override
    public MqSendResult sendDelay(MqMsg msg) {
        try {
            Message rocketmq = msgBuilder.build(msg, encoder);
            rocketmq.setWaitStoreMsgOK(true);

            SendResult result = producer.send(rocketmq);
            return sendResultBuilder.build(result);
        } catch (Exception e) {
            throw new MqMsgSendException(msg.getTopic(), e);
        }
    }

    @Override
    public void asyncSend(MqMsg msg, MqSendCallback cb) {
        try {
            Message rocketmq = msgBuilder.build(msg, encoder);
            producer.send(rocketmq, buildCallback(cb));
        } catch (Exception e) {
            throw new MqMsgSendException(msg.getTopic(), e);
        }
    }

    @Override
    public void asyncSendOrderly(MqMsg msg, MqSendCallback cb) {
        try {
            Message rocketmq = msgBuilder.build(msg, encoder);
            producer.send(rocketmq, new MqMsgSelector(), msg.getShardingKey(), buildCallback(cb));
        } catch (Exception e) {
            throw new MqMsgSendException(msg.getTopic(), e);
        }
    }

    @Override
    public void shutdown() {
        producer.shutdown();
    }

    private SendCallback buildCallback(MqSendCallback cb) {
        return new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                MqSendResult mqSendResult = sendResultBuilder.build(sendResult);
                cb.onSuccess(mqSendResult);
            }

            @Override
            public void onException(Throwable e) {
                cb.onFailure(e);
            }
        };
    }
}
